package com.king.app.function

import java.sql.{Connection, DriverManager, PreparedStatement, SQLException, Statement}

import com.alibaba.fastjson.JSONObject
import com.king.common.GmallConfig
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.function._
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}

/**
 * @Author: KingWang
 * @Date: 2022/1/26  
 * @Desc:
 **/
class DimSinkFunction extends RichSinkFunction[JSONObject]{

  var conn:Connection = _
  var stmt:Statement = _
  override def open(parameters: Configuration): Unit = {

    try {
//      Class.forName(GmallConfig.PHOENIX_DRIVER)   添加这行会报错
      conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER)
      conn.setSchema(GmallConfig.HBASE_SCHEMA)
    }catch{
      case e:ClassNotFoundException => println("org.apache.phoenix.jdbc.PhoenixDriver未找到")
      case _:Exception => println("phoenix驱动加载异常")
    }

  }

  //value:{"database":"gmall-210325-realtime","before":{},"after":{"operate_type":"insert","sink_type":"kafka","sink_table":"dwd_order_info","source_table":"order_info","sink_columns":"id,user_id"},"type":"read","table":"table_process","ts":1643303146023}
  override def invoke(value: JSONObject, context: SinkFunction.Context): Unit = {

    val sinkTableSchema = GmallConfig.HBASE_SCHEMA
    val sinkTable = value.getString("sinkTable")
    val recordObj = value.getJSONObject("after")

    try{
      //UPSERT INTO TEST(NAME,ID) VALUES('foo',123);
      val fields = recordObj.keySet().toArray.mkString(",")
      val values = recordObj.values().toArray.mkString("','")
      val upsertSQL = s"upsert into $sinkTableSchema.$sinkTable($fields) values('$values') "
//      println("写入Hbase upsertSQL>>>>" +upsertSQL)
      stmt = conn.createStatement()
      stmt.executeUpdate(upsertSQL)
      conn.commit()   //提交一定要有
    }catch {
      case e:SQLException => e.printStackTrace()
    }finally {
      if(null != stmt)  stmt.close()
    }



  }

  override def close(): Unit = {
    conn.close()
  }

}
